package com.bdqn.mr.rj;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

public class ReduceJoinReducer extends Reducer<Text, TableBean, TableBean, NullWritable> {
    /**
     * @param key     产品的编号
     * @param values  两张表数据的合表
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {
        // 01	1001	1	order
        // 01	1004	4	order
        // 01	小米		pd

        // 重新定义一个变量，保存pd的一条数据
        TableBean tempPd = new TableBean();

        // 定义一个列表，保存所有的order数据
        List<TableBean> orders = new ArrayList<>();

        for (TableBean bean : values) {
            if (bean.getTableName().equals("order")) {
                // order
                // 这样，对吗
                try {
                    TableBean order = new TableBean();
                    BeanUtils.copyProperties(order,bean);
                    orders.add(order);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }
//                order.setOrderId(bean.getOrderId());
//                order.setPid(bean.getPid());
//                order.setAmount(bean.getAmount());
//                order.setpName(bean.getpName());
//                order.setTableName(bean.getTableName());


            } else {
                // pd
                // 这样行不行？

                // 这样就可以了。
                // 如果这里有20个属性，或者100多个，
//                tempPd.setPid(bean.getPid());
//                tempPd.setpName(bean.getpName());
//                tempPd.setTableName(bean.getTableName());

                // 目的：把一个对象中的属性，复制到另外一个对象中去

                try {
                    BeanUtils.copyProperties(tempPd,bean);
                } catch (IllegalAccessException e) {
                    e.printStackTrace();
                } catch (InvocationTargetException e) {
                    e.printStackTrace();
                }

            }
        }

        // 把 pd 的数据，赋值到 orders 里面去
        for (TableBean order : orders) {
            order.setpName(tempPd.getpName());
            context.write(order, NullWritable.get());
        }

    }
}
